package cn.ac.iie.bdb_to_mongo.mongohandler;

import cn.ac.iie.bdb_to_mongo.BdbMain;
import cn.ac.iie.bdb_to_mongo.commons.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import com.sleepycat.je.*;
import java.math.BigInteger;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @auther Fighter Created on 2016/12/2.
 */
public class UpdateTask implements Runnable {

	private Cursor cursor;
	private long sliceLen;
	private boolean isLastThread;
	private HashMap<String, Object> updateValues = new LinkedHashMap<>();//保持读写顺序一致
	private final Logger log = LoggerFactory.getLogger(UpdateTask.class);
	private int i;

	private Set<String> allIds = new HashSet<>();

	private static String[] queryFields = {"_id", "ct", "wx_ic"};

	UpdateTask(int i, long sliceLen, boolean isLastThread) {
		this.i = i;
		this.sliceLen = sliceLen;
		this.isLastThread = isLastThread;
	}

	/**
	 * 实现更新mongo操作
	 */
	@Override
	public void run() {

		try {

			this.cursor = getSpecCursor(i, sliceLen);

			DatabaseEntry foundKey = new DatabaseEntry();
			DatabaseEntry foundData = new DatabaseEntry();

			if (isLastThread) {
				//最后一个线程则处理剩余所有
				while (cursor.getNext(foundKey, foundData, LockMode.READ_UNCOMMITTED) == OperationStatus.SUCCESS) {
					String keyString = new String(foundKey.getData(), "UTF-8");
					String dataString = new String(foundData.getData(), "UTF-8");
					//更新mongo
					updateValues.put(keyString, Long.parseLong(dataString));
					allIds.add(keyString);
				}
			} else {
				//非最后一个线程则处理定量的bdb数据分段
				for (int i = 0; i < sliceLen; i++) {
					if (cursor.getNext(foundKey, foundData, LockMode.READ_UNCOMMITTED) == OperationStatus.SUCCESS) {
						String keyString = new String(foundKey.getData(), "UTF-8");
						String dataString = new String(foundData.getData(), "UTF-8");
						updateValues.put(keyString, Long.parseLong(dataString));
						allIds.add(keyString);
					}
				}
			}

			//添加不为空判断,不为空才能更新mongo和发送
			if (updateValues.size() > 0) {
				/*try {
                    for (Map.Entry<String, Object> entry : updateValues.entrySet()) {
//                        Object[] obj = new Object[7];
                        Object[] obj = new Object[8];
                        obj[0] = entry.getKey();//_id
                        obj[1] = entry.getKey().split("_")[2];//_key
//                        obj[2] = "WeChat";//asp obj[2]更改为ct
                        obj[3] = "M-WX02-A";//spec
                        obj[4] = "40";//_ch
                        obj[5] = entry.getValue();//wx_ic
//                        obj[6] = System.currentTimeMillis() / 1000;
                        obj[7]=string2MD5(obj[1].toString());//wx_key_fng 32位md5
                        datas.add(obj);
                    }
                } catch (Exception e) {
                    log.error("add msg data error!" + ExceptionUtils.getFullStackTrace(e));
                }*/

				try {
					Long ut1 = System.currentTimeMillis() / 1000;
					//更新mongo旧表docinfo
					UpdateMongoTask.mongoHelper.update(
							BdbMain.configuration.getString(Constants.MONGO_UPDATE_KEYCOL),
							BdbMain.configuration.getString(Constants.MONGO_UPDATE_UPDATECOL),
							ut1,
							updateValues
					);

					//查询ct, wx_ic,_id
					List<HashMap<String, Object>> resList1 = UpdateMongoTask.mongoHelper.find(
							BdbMain.configuration.getString(Constants.MONGO_UPDATE_KEYCOL),
							updateValues.keySet(),
							queryFields
					);

					//更新旧mongo成功就发送消息队列
					if (resList1 != null && !resList1.isEmpty()) {
						addMsg(BdbMain.toMqQueue1, ut1, resList1);
					}
					/*//取差集回写mq
                    reWrite(resList1);*/
					log.info("update old table " + updateValues.size() + " records");
				} catch (Exception e) {

					log.error("update mongo error!" + ExceptionUtils.getFullStackTrace(e));
				}

				try {
					Long ut2 = System.currentTimeMillis() / 1000;
					//更新mongo新表ivrmsg
					UpdateMongoTask.mongoHelper1.update(
							BdbMain.configuration.getString(Constants.MONGO_UPDATE_KEYCOL),
							BdbMain.configuration.getString(Constants.MONGO_UPDATE_UPDATECOL),
							ut2,
							updateValues
					);

					//查询ct, wx_ic
					List<HashMap<String, Object>> resList2 = UpdateMongoTask.mongoHelper.find(
							BdbMain.configuration.getString(Constants.MONGO_UPDATE_KEYCOL),
							updateValues.keySet(),
							queryFields
					);
					if (resList2 != null && !resList2.isEmpty()) {
						addMsg(BdbMain.toMqQueue2, ut2, resList2);
						reWrite(resList2);//以新库为准
					}
					log.info("update new table " + updateValues.size() + " records");
				} catch (Exception e) {
					log.error("update mongo error!" + ExceptionUtils.getFullStackTrace(e));
				}
			}
		} catch (Exception e) {
			log.error("read bdb error " + ExceptionUtils.getFullStackTrace(e));
		} finally {
			updateValues.clear();
			allIds.clear();
			cursor.close();
			cursor = null;
		}
	}

	/**
	 * 给每个线程提供特定位置的游标
	 *
	 * @param i 第i个线程 0,1,...,threadNum-1
	 * @param sliceLen 每个线程负责bdb记录数
	 * @return 返回特定位置游标
	 */
	public Cursor getSpecCursor(int i, long sliceLen) {

		CursorConfig config = new CursorConfig();
		config.setReadUncommitted(true);
//        Cursor myCursor = BdbMain.testDB.getDb().openCursor(null, config);
		Cursor myCursor = BdbMain.testDB.getCursor();

		//bdb第一个分段的游标
		if (0 == i) {
			return myCursor;
		}
		//bdb第二个以及以后分段的游标

		//先定位到bdb第一条记录
		if (myCursor.getNext(new DatabaseEntry(), new DatabaseEntry(), LockMode.DEFAULT) == OperationStatus.SUCCESS) {
			//游标指向该片前一位置
			myCursor.skipNext(i * sliceLen - 1, new DatabaseEntry(), new DatabaseEntry(), LockMode.DEFAULT);
			return myCursor;
		} else {
			log.error("get Cursor error!");
			return null;
		}
	}

	//发送数据
	public void addMsg(LinkedBlockingDeque<REMessage> queue, Long ut, List<HashMap<String, Object>> resList) {
//        String[] colName = {"_id", "_key", "asp", "spec", "_ch", "wx_ic", "ut","wx_key_fng"};

		String[] colName = {"_id", "_key", "ct", "spec", "_ch", "wx_ic", "ut", "wx_key_fng"};
		REFieldType[] types = {REFieldType.String, REFieldType.String, REFieldType.Long, REFieldType.String,
			REFieldType.Long, REFieldType.Long, REFieldType.Long, REFieldType.String};

		List<Object[]> datas = new ArrayList<>();

		for (int j = 0; j < resList.size(); j++) {
			Object[] obj = new Object[8];
			obj[0] = resList.get(j).get("_id");//_id
			obj[1] = obj[0].toString().split("_")[2];//_key
			obj[2] = resList.get(j).get("ct");
			obj[3] = "M-WX02-A";//spec
			obj[4] = "40";//_ch
			obj[5] = resList.get(j).get("wx_ic");
			obj[6] = ut;
			obj[7] = string2MD5(obj[1].toString());//wx_key_fng 32位md5
			datas.add(obj);
		}

		REMessage msg = new REMessage(colName, types, datas, null, null);
		boolean flag = queue.offer(msg);
		while (!flag) {
			flag = queue.offer(msg);
		}
	}

	public void reWrite(List<HashMap<String, Object>> resList) {
		//获取id差集
		Set<String> ids = new HashSet<>(allIds);
		for (HashMap<String, Object> m : resList) {
			ids.remove(m.get("_id"));
		}

		String[] colNames = {"_id", "count"};
		REFieldType[] types = {REFieldType.String, REFieldType.Long};

		List<Object[]> datas = new ArrayList<>();
		for (String s : ids) {
			Object[] obj = new Object[2];
			obj[0] = s;
			obj[1] = updateValues.get(s);
			datas.add(obj);
		}
		REMessage msg = new REMessage(colNames, types, datas, null, null);
		try {
			BdbMain.reWriteQueue.put(msg);
		} catch (InterruptedException e) {
			log.error("put in reWriteQueue error! {}", ExceptionUtils.getFullStackTrace(e));
		}
	}

	public String string2MD5(String plaintext) {
		MessageDigest m;
		try {
			m = MessageDigest.getInstance("MD5");
		} catch (NoSuchAlgorithmException e) {
			System.out.println(e.toString());
			e.printStackTrace();
			return "";
		}
		m.reset();
		m.update(plaintext.getBytes());
		byte[] digest = m.digest();
		BigInteger bigInt = new BigInteger(1, digest);
		String hashtext = bigInt.toString(16);
		while (hashtext.length() < 32) {
			hashtext = "0" + hashtext;
		}
		return hashtext;
	}
}
